package com.example.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;

@Component
public class DemoConsumerB {

    private Logger logger = LoggerFactory.getLogger(getClass());
    private AtomicInteger times = new AtomicInteger(0);
    @KafkaListener(topics = "topic1",groupId = "group-B")
    public void onMessage(ConsumerRecord<Integer, String> record,Acknowledgment acknowledgment) {
        if (times.get() % 2 == 0) {
            acknowledgment.nack(0);
            logger.error("消费异常");
        } else {
            logger.info("[onMessage][线程编号:{},partition:{} 消息内容：{}]", Thread.currentThread().getId(), record.partition(), record);
            acknowledgment.acknowledge();
        }
        times.incrementAndGet();

    }

}
